package cn.doitedu.day07

import org.apache.spark.util.AccumulatorV2
import scala.collection._

/**
 * @Date 22.4.6
 * @Created by HANGGE
 * @Description 自定义累加器  统计单词的次数
 *             AccumulatorV2[IN , OUT]
 *             IN  输入的数据类型   add(1)   Long   add("a") String
 *             OUT 输出数据类型     累加器.value  100 Long   {"a" , 23}  Map
 */
class MyAccumulator extends  AccumulatorV2[String , Map[String , Int]]{

  // 封装返回的结果    可变的Map
    val mp = mutable.Map[String,Int]()
  /**
   * 是否是一个新的累加
   * @return
   */
  override def isZero: Boolean = mp.isEmpty

  /**
   * 实例化使用
   *   自定义的累加器要注册   调用copy方法注册当前自定义的累加器
   * @return
   */
  override def copy(): AccumulatorV2[String, Map[String, Int]] = {
    new  MyAccumulator()
  }
  /**
   * 累加器 清空
   */
  override def reset(): Unit = mp.clear()

  /**
   * 每个Executor的计算逻辑
   *   使用Map集合的特点  统计单词的次数
   * @param v
   */
  override def add(v: String): Unit = {
    val cnt = mp.getOrElse(v, 0)
    mp.update(v , cnt+1)
  }

  /**
   * 在Driver 汇总结果
   * @param other
   *              reduce函数  (x1,x2)  ==> x1
   */
  override def merge(other: AccumulatorV2[String, Map[String, Int]]): Unit = {
    var mp1 = mp            // (a,11) (b,12) (d,10)
    var mp2 = other.value    // (a,21) (d,10) (e,21)
    // 遍历集合二  mp2 将数据同步到mp1中
    mp2.foreach(mp=>{
      val word = mp._1
      val cnt = mp._2
      val cnt2 = mp1.getOrElse(word, 0)
      // 汇总结果
      mp1.update(word , cnt2+cnt)
    })
  }

  /**
   * 返回最终结果的方法
   * @return
   */
  override def value: Map[String, Int] = mp
}
